package com.xueyuan.wata.daph.node.spark3.dataframe.stream.connector.common

import com.xueyuan.wata.daph.spark3.api.node.dataframe.connector.input.DataFrameInput
import org.apache.spark.sql.DataFrame

class CommonInput extends DataFrameInput {
  override protected def in(): DataFrame = {
    val config = nodeConfig.asInstanceOf[CommonInputConfig]
    val format = config.format
    val cfg = config.cfg
    val path = config.path

    val res = spark.readStream
      .format(format)
      .options(cfg)
      .load(path)
    res
  }

  override def getNodeConfigClass = classOf[CommonInputConfig]
}